// Description: 描述
package mq

import (
	"errors"
	"fmt"

	"github.com/zeromicro/go-zero/core/logx"

	"gitlab.vspncd.com/mic/common/mq/internal/kq"
)

type (
	KafkaMq interface {
		// 开启
		Start()
		// 关闭
		Stop()
		// 设置链路
		SetUp(adds []string, topic string)
		// 注册消费者
		RegConsumer(topic string, group string, handle func(key, value string) error)
		// 发送
		Send(topic string, v interface{}) error
		// 指定key发送
		SendWithKey(topic string, key string, v interface{}) error
	}

	defaultKafkaMq struct {
		mode  string // 启动模式
		kafka map[string]kq.TargetKafka
	}
)

func NewKafkaMq(mode ...string) KafkaMq {
	target := &defaultKafkaMq{
		kafka: map[string]kq.TargetKafka{},
	}

	if len(mode) > 0 {
		target.mode = mode[0]
	}

	return target
}

func (k *defaultKafkaMq) Start() {
	for _, q := range k.kafka {
		q.Start()
	}
}

func (k *defaultKafkaMq) Stop() {
	for _, q := range k.kafka {
		q.Stop()
	}
}

// 初始化时使用，程序中不要调用
func (k *defaultKafkaMq) SetUp(adds []string, topic string) {
	kafka := kq.NewTargetKafka(adds, topic, kq.WithMode(k.mode))
	k.kafka[topic] = kafka
}

// 注册kafka消费端
func (k *defaultKafkaMq) RegConsumer(topic string, group string, handle func(key, value string) error) {
	kafka, ok := k.kafka[topic]
	if ok {
		kafka.RegConsumer(group, handle)
	}
}

// 发送kafka消息
func (k *defaultKafkaMq) Send(topic string, v interface{}) (err error) {
	kafka, ok := k.kafka[topic]
	if !ok {
		panic(fmt.Errorf("未知的topic[%s]", topic))
	}

	err = kafka.Send(v)
	if err != nil {
		logx.Error("send msg error", err)
		return errors.New(fmt.Sprintf("kafka消息推送失败[%v]", v))
	}

	return
}

func (k *defaultKafkaMq) SendWithKey(topic string, key string, v interface{}) (err error) {
	kafka, ok := k.kafka[topic]
	if !ok {
		panic(fmt.Errorf("未知的topic[%s]", topic))
	}

	err = kafka.SendWithKey(key, v)
	if err != nil {
		logx.Error("send msg error", err)
		return errors.New(fmt.Sprintf("kafka消息推送失败[%v]", v))
	}

	return
}
